package com.book.main;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.book.util.ElasticsearchUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class Consumption {
	private static Logger log = LoggerFactory.getLogger(Consumption.class);

	private final static String QUEUE_NAME = "book";
	public static void main(String[] args) throws IOException, TimeoutException {
			log.info("==start==");
		    ExecutorService executors = Executors.newFixedThreadPool(64);

		    //定时任务计数器
			AtomicInteger count = new AtomicInteger(0);
			Vector<Integer> vector = new Vector<Integer>();

			Timer timer = new Timer();
			Timer timer2 = new Timer();
			timer.schedule(new TimerTask() {
				@Override
				public void run() {
//					log.info(count.get()+"");
					vector.add(count.get());
					count.set(0);
				}
			},1000,1000);
			timer2.schedule(new TimerTask() {
				@Override
				public void run() {
					log.info("--------"+Collections.max(vector));
					vector.clear();
				}
			},10000,10000);

			ConnectionFactory factory = new ConnectionFactory();
			factory.setHost("127.0.0.1");
			factory.setPort(5672);
			factory.setUsername("guest");
			factory.setPassword("guest");
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
			channel.queueDeclare(QUEUE_NAME, true, false, false, null);
			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
					String message = new String(body, "UTF-8");
					//添加数据
//					count.incrementAndGet();
//					int status = ElasticsearchUtil.index("book", "book", UUID.randomUUID().toString().replaceAll("-", ""),message);
//					log.info(count.get()+"");
					executors.execute(new SaveBook(count,message,Thread.currentThread().getName()));
				}
			};
			channel.basicConsume(QUEUE_NAME, true, consumer);
	}
}

class SaveBook implements Runnable{
	private static Logger log = LoggerFactory.getLogger(SaveBook.class);
	AtomicInteger count;
	String msg;
	String threadName;

	public SaveBook(AtomicInteger count,String msg,String threadName) {
		this.count = count;
		this.msg = msg;
		this.threadName =threadName;
	}

	public String getThreadName() {
		return threadName;
	}

	public void setThreadName(String threadName) {
		this.threadName = threadName;
	}

	public String getMsg() {
		return msg;
	}

	public void setMsg(String msg) {
		this.msg = msg;
	}

	public AtomicInteger getCount() {
		return count;
	}

	public void setCount(AtomicInteger count) {
		this.count = count;
	}

	@Override
	public void run() {
//		log.info("threadName:"+threadName);
		synchronized (SaveBook.class){
			count.incrementAndGet();
			int status = ElasticsearchUtil.index("book", "book", UUID.randomUUID().toString().replaceAll("-", ""),msg);
//			log.info("----"+count.get());
		}

	}
}